package test.consumer;

import javax.jms.*;

import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.prm.excelserver.protobuf.*;

import java.util.concurrent.*;

public class RxThread implements Runnable {
	private MessageConsumer myConsumer;
	static private boolean isRunning;
	private String dataKind;
	private final CountDownLatch startsignal;
	private final CountDownLatch donesignal;
	private ExcelServer server;
    private String[] instItems = new String[] {"feedcode", "strike", "isCall", "best_bid", "best_ask"};
    private String[] tradeItems = new String[] {"feedcode", "isBuy", "price", "volume"};

	public RxThread(MessageConsumer consumer, String dataKind, CountDownLatch startsignal, CountDownLatch donesignal) {
		this.myConsumer = consumer;
		this.dataKind = dataKind;
		this.startsignal = startsignal;
		this.donesignal = donesignal;
	}
	
	public void run() {
		isRunning = true;
		try {
			startsignal.await();
		} catch (InterruptedException e) {
			System.out.println("CountDownLatch exception error");
        }
		System.out.println("Received Start Signal from the main class.. blocking in receive function");
		CountDownLatch latch = new CountDownLatch(1);

		if (dataKind == "Instrument") {
			server = new ExcelServer("prm", "Instrument", instItems, latch);
		} else if (dataKind == "Trade") {
			server = new ExcelServer("prm", "Trade", tradeItems, latch);
		}
        server.ActivateServer();
		try {
			latch.await();
		} catch(InterruptedException e) {}
    	while(isRunning) {
			try {
				Message message = myConsumer.receive();
				System.out.println("Received Message from the activemq");
		        if (message instanceof BytesMessage) {
		        	BytesMessage byteMessage = (BytesMessage) message;
		        	
		        	byte[] newBuf = new byte[(int)byteMessage.getBodyLength()];
		        	byteMessage.readBytes(newBuf);
		            
		            try {
			            if (dataKind == "Instrument") {
			            	ExcelServerProtobuf.Instrument inst = ExcelServerProtobuf.Instrument.parseFrom(newBuf);
			            	System.out.println("Received message:");
			            	System.out.println(inst.toString());
			            	server.setInstMsg(inst);
			            	server.runPostAdviseAllItems(dataKind);			            	
			            } else if (dataKind == "Trade") {
			            	ExcelServerProtobuf.trade tradeMsg = ExcelServerProtobuf.trade.parseFrom(newBuf);
			            	System.out.println("Received message:");
			            	System.out.println(tradeMsg.toString());
			            	server.setTradeMsg(tradeMsg);
			            	server.runPostAdviseAllItems(dataKind);
			            }
			            

		            } catch (InvalidProtocolBufferException e) {
		            	System.out.println("protobuf parsing error");
		            }
		        }

			} catch (JMSException e) {
				System.out.println(e.toString());
			}
		}
    	donesignal.countDown();
	}
	
	static public void stopRunning() {
		isRunning = false;
	}
}
